package com.zx.learn.flink.sink;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

/**
 * 打印sink.
 *
 * @author Ken.Zheng
 * @date 2022/11/29
 **/
@Slf4j
public class SelfDefinePrintSink<IN> extends RichSinkFunction<IN> {

    private int indexOfThisSubtask;

    @Override
    public void open(Configuration parameters) throws Exception {
        RuntimeContext runtimeContext = getRuntimeContext();
        indexOfThisSubtask = runtimeContext.getIndexOfThisSubtask();
    }

    @Override
    public void invoke(IN value, Context context) {
        log.info("自定义输出{}:{}", (indexOfThisSubtask + 1), value);
    }
}
